package com.lhl.mq.mqServer.core;

import com.lhl.mq.common.Consumer;
import com.lhl.mq.common.ConsumerEnv;
import com.lhl.mq.common.MqException;
import com.lhl.mq.mqServer.VirtualHost;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 通过这个类来实现消费消息的核心逻辑
 */

public class ConsumerManage {
    //持有上层的 virtualHost 对象的引用，用来操作数据
    private VirtualHost prent;
    //创建一个线程池用去来执行具体的回调方法
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    //创建一个阻塞队列来存放令牌
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    //创建一个扫描线程
    private Thread scannerThread = null;

    public ConsumerManage(VirtualHost prent) {
        this.prent = prent;

        scannerThread = new Thread(() ->{
            //扫描线程一直扫描阻塞队列，看哪个队列有消息需要消费
            while (true){
                try {
                    //拿到令牌
                    String queueName = tokenQueue.take();
                    //找到队列
                    MSGQueue queue = prent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null){
                        //如果队列为空，则抛出异常
                        throw new MqException("[ConsumerManger] 扫描失败，队列不存在,queueName = " + queueName);
                    }
                    //根据令牌消费一个消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    throw new RuntimeException(e);
                }

            }

        });
        //将线程设为后台线程
        scannerThread.setDaemon(true);
        //关闭线程
        scannerThread.start();

    }

    //这个方法在发送消息时被触发
    public void  notifyConsume(String token) throws InterruptedException {
        //将队列的名字当作令牌，存入阻塞队列里面
        tokenQueue.put(token);
    }

    //添加消费者到被订阅的队列里面
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //查询队列
        MSGQueue queue = prent.getMemoryDataCenter().getQueue(queueName);
        //若队列不存在，抛出异常
        if (queue == null){
            throw new MqException("[ConsumerManger] 添加消费者失败，队列不存在，queueName = " + queueName );
        }
        //构造一个消费者
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            //将消费者添加到队列中
            queue.addConsumerEnv(consumerEnv);

            //如果此时队列里面有消息需要立即消费掉
            int messageCount = prent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < messageCount; i++) {
                //该方法调用一次就消费一个方法
                consumeMessage(queue);
            }
        }

    }

    //消费消息
    private void consumeMessage(MSGQueue queue) {
        //选出一个消费者来消费消息
        ConsumerEnv consumerEnv = queue.chooseConsumer();
        if (consumerEnv == null){
            //说明此时该队列没有消费者订阅
            return;
        }
        //从队列中取出一个消息
        Message message = prent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null){
            //说明此时队列里面还没有消息
            return;
        }
        //把消息带入到消费者的回调方法中
        workerPool.submit(() -> {
            try {
                //执行回调之前，先把要进行消费的消息加入到待确认的集合中
                prent.getMemoryDataCenter().addQueueMessageWaitAckMap(queue.getName(),message);
                //执行回调操作（当回调方法正确执行完，则表明消息是被正确消费掉了）
                consumerEnv.getConsumer().handleDelivery(consumerEnv.getConsumerTag(),message.getBasicProperties(),message.getBody());
                //如果消费者的autoAck的类型为true（自动应答）,则认为消息已经被正确的消费掉了
                //如果autoAck为false，则暂时不处理，交给消费者后续手动调用
                if (consumerEnv.isAutoAck()){
                    //删除硬盘中的消息
                    if (message.getDeliverMode() == 2) {
                        prent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    //删除内存中的消息
                    prent.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    //删除待确认集合中的消息
                    prent.getMemoryDataCenter().deleteQueueMessageWaitAckMap(queue.getName(), message.getMessageId());
                    System.out.println("[ConsumerManger] 成功消费消息，queueName = " + queue.getName());
                }

            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });

    }
}
